package function;

import bean.TrafficEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import org.apache.flink.api.common.functions.RichMapFunction;

public class TrafficMapFunction extends RichMapFunction<String, TrafficEvent> {
    @Override
    public TrafficEvent map(String value) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        TrafficEvent trafficEvent = new TrafficEvent();
        trafficEvent.setEventId(jsonObject.getString("eventId"));
        trafficEvent.setCameraId(jsonObject.getString("cameraId"));
        trafficEvent.setLicensePlate(jsonObject.getString("licensePlate"));
        trafficEvent.setVehicleType(jsonObject.getString("vehicleType"));
        trafficEvent.setRoadId(jsonObject.getString("roadId"));
        trafficEvent.setCrossingId(jsonObject.getString("crossingId"));
        trafficEvent.setTimestamp(jsonObject.getLong("ts"));
        trafficEvent.setSpeed(jsonObject.getDouble("speed"));
        trafficEvent.setDirection(jsonObject.getString("direction"));
        trafficEvent.setLane(jsonObject.getInteger("lane"));
        trafficEvent.setEventType(jsonObject.getString("eventType"));
        trafficEvent.setAreaId(jsonObject.getString("areaId"));
        return trafficEvent;
    }
}
